package org.helium.redis.widgets.redis.client;

import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
import org.helium.util.Combo3;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import redis.clients.jedis.HostAndPort;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Protocol;
import redis.clients.jedis.exceptions.JedisConnectionException;
import redis.clients.jedis.exceptions.JedisException;
import redis.clients.util.Pool;

import java.util.Arrays;
import java.util.List;
import java.util.Set;
import java.util.concurrent.ConcurrentLinkedQueue;

/**
 * Created by yibo on 2017-6-9.
 */
public class JedisSentinelPoolFeinno extends Pool<Jedis> {

    protected GenericObjectPoolConfig poolConfig;

    protected int timeout = Protocol.DEFAULT_TIMEOUT;

    protected String password;

    protected int database = Protocol.DEFAULT_DATABASE;

    // protected Set<MasterListener> masterListeners = new
    // HashSet<MasterListener>();

    protected static Logger log = LoggerFactory.getLogger(JedisSentinelPoolFeinno.class);

    public JedisSentinelPoolFeinno(String masterName, Set<String> sentinels, final GenericObjectPoolConfig poolConfig) {
        this(masterName, sentinels, poolConfig, Protocol.DEFAULT_TIMEOUT, null, Protocol.DEFAULT_DATABASE);
    }

    public JedisSentinelPoolFeinno(String masterName, Set<String> sentinels) {
        this(masterName, sentinels, new GenericObjectPoolConfig(), Protocol.DEFAULT_TIMEOUT, null,
                Protocol.DEFAULT_DATABASE);
    }

    public JedisSentinelPoolFeinno(String masterName, Set<String> sentinels, String password) {
        this(masterName, sentinels, new GenericObjectPoolConfig(), Protocol.DEFAULT_TIMEOUT, password);
    }

    public JedisSentinelPoolFeinno(String masterName, Set<String> sentinels, final GenericObjectPoolConfig poolConfig,
                                   int timeout, final String password) {
        this(masterName, sentinels, poolConfig, timeout, password, Protocol.DEFAULT_DATABASE);
    }

    public JedisSentinelPoolFeinno(String masterName, Set<String> sentinels, final GenericObjectPoolConfig poolConfig,
                                   final int timeout) {
        this(masterName, sentinels, poolConfig, timeout, null, Protocol.DEFAULT_DATABASE);
    }

    public JedisSentinelPoolFeinno(String masterName, Set<String> sentinels, final GenericObjectPoolConfig poolConfig,
                                   final String password) {
        this(masterName, sentinels, poolConfig, Protocol.DEFAULT_TIMEOUT, password);
    }

    public JedisSentinelPoolFeinno(String masterName, Set<String> sentinels, final GenericObjectPoolConfig poolConfig,
                                   int timeout, final String password, final int database) {
        this.poolConfig = poolConfig;
        this.timeout = timeout;
        this.password = password;
        this.database = database;

        // HostAndPort master = initSentinels(sentinels, masterName);
        // initPool(master);
        refreshMaster(masterName,sentinels);
        SentinelManager.addEntry(masterName, sentinels, this);
    }

    private volatile JedisFactory factory;
    private volatile HostAndPort currentHostMaster;

    public void destroy() {
        // for (MasterListener m : masterListeners) {
        // m.shutdown();
        // }

        super.destroy();
    }

    public HostAndPort getCurrentHostMaster() {
        return currentHostMaster;
    }

    public void refreshMaster(String masterName, Set<String> sentinels) {
        HostAndPort master = null;
        boolean sentinelAvailable = false;
        for (String sentinel : sentinels) {
            final HostAndPort hap = toHostAndPort(Arrays.asList(sentinel.split(":")));
            Jedis jedis = null;
            try {
                jedis = new Jedis(hap.getHost(), hap.getPort());
                List<String> masterAddr = jedis.sentinelGetMasterAddrByName(masterName);
                // connected to sentinel...
                sentinelAvailable = true;
                if (masterAddr == null || masterAddr.size() != 2) {
                    log.error("Can not get master addr, master name: " + masterName + ". Sentinel: " + hap + ".");
                    continue;
                }
                master = toHostAndPort(masterAddr);
//				log.info("Found Redis master at " + master);
                break;
            } catch (JedisConnectionException e) {
                log.warn("Cannot connect to sentinel running @ " + hap + ". Trying next one.");
            } finally {
                if (jedis != null) {
                    jedis.close();
                }
            }
        }

        if (master == null) {
            if (sentinelAvailable) {
                currentHostMaster = null;
                // can connect to sentinel, but master name seems to not
                // monitored
                log.error("sentinel is Available, but " + masterName + " seems to be not monitored...");
            } else {
                log.error("All sentinels down, cannot determine where is " + masterName
                        + " master is running...");
            }

        }else
            initPool(master);

    }

    private void initPool(HostAndPort master) {
        if (master!=null&&!master.equals(currentHostMaster)) {
            currentHostMaster = master;
            if (factory == null) {
                factory = new JedisFactory(master.getHost(), master.getPort(), timeout, password, database);
                initPool(poolConfig, factory);
            } else {
                factory.setHostAndPort(currentHostMaster);
                // although we clear the pool, we still have to check the
                // returned object
                // in getResource, this call only clears idle instances, not
                // borrowed instances
                internalPool.clear();
            }
            log.info("Created JedisPool to master at " + master);
        }
    }

    private HostAndPort initSentinels(Set<String> sentinels, final String masterName) {

        HostAndPort master = null;
        boolean sentinelAvailable = false;

        log.info("Trying to find master from available Sentinels...");

        for (String sentinel : sentinels) {
            final HostAndPort hap = toHostAndPort(Arrays.asList(sentinel.split(":")));

            log.info("Connecting to Sentinel " + hap);

            Jedis jedis = null;
            try {
                jedis = new Jedis(hap.getHost(), hap.getPort());

                List<String> masterAddr = jedis.sentinelGetMasterAddrByName(masterName);

                // connected to sentinel...
                sentinelAvailable = true;

                if (masterAddr == null || masterAddr.size() != 2) {
                    log.warn("Can not get master addr, master name: " + masterName + ". Sentinel: " + hap + ".");
                    continue;
                }

                master = toHostAndPort(masterAddr);
                log.info("Found Redis master at " + master);
                break;
            } catch (JedisConnectionException e) {
                log.warn("Cannot connect to sentinel running @ " + hap + ". Trying next one.");
            } finally {
                if (jedis != null) {
                    jedis.close();
                }
            }
        }

        if (master == null) {
            if (sentinelAvailable) {
                // can connect to sentinel, but master name seems to not
                // monitored
                throw new JedisException("Can connect to sentinel, but " + masterName + " seems to be not monitored...");
            } else {
                throw new JedisConnectionException("All sentinels down, cannot determine where is " + masterName
                        + " master is running...");
            }
        }

        log.info("Redis master running at " + master + ", starting Sentinel listeners...");

        // for (String sentinel : sentinels) {
        // final HostAndPort hap =
        // toHostAndPort(Arrays.asList(sentinel.split(":")));
        // MasterListener masterListener = new MasterListener(masterName,
        // hap.getHost(), hap.getPort());
        // masterListeners.add(masterListener);
        // masterListener.start();
        // }

        return master;
    }

    private HostAndPort toHostAndPort(List<String> getMasterAddrByNameResult) {
        String host = getMasterAddrByNameResult.get(0);
        int port = Integer.parseInt(getMasterAddrByNameResult.get(1));

        return new HostAndPort(host, port);
    }

    @Override
    public Jedis getResource() {

        if(currentHostMaster==null)
            return null;
        while (true) {
            Jedis jedis = super.getResource();
            jedis.setDataSource(this);

            // get a reference because it can change concurrently
            final HostAndPort master = currentHostMaster;
            final HostAndPort connection = new HostAndPort(jedis.getClient().getHost(), jedis.getClient().getPort());

            if (master.equals(connection)) {
                // connected to the correct master
                return jedis;
            } else {
                returnBrokenResource(jedis);
            }
        }
    }

    public void returnBrokenResource(final Jedis resource) {
        if (resource != null) {
            returnBrokenResourceObject(resource);
        }
    }

    public void returnResource(final Jedis resource) {
        if (resource != null) {
            resource.resetState();
            returnResourceObject(resource);
        }
    }

    static class SentinelManager {
        private static Thread masterLoopThread;
        private static ConcurrentLinkedQueue<Combo3<String, Set<String>, JedisSentinelPoolFeinno>> sentinelGroups;

        static {
            sentinelGroups = new ConcurrentLinkedQueue<Combo3<String, Set<String>, JedisSentinelPoolFeinno>>();
            masterLoopThread = new Thread(new Runnable() {
                @Override
                public void run() {
                    checkSentinel();
                }
            }, "SentinelmasterLoopThread");
            masterLoopThread.setDaemon(true);
            masterLoopThread.start();
        }

        static void addEntry(String masterName, Set<String> sentinels, JedisSentinelPoolFeinno instance) {
            sentinelGroups.offer(new Combo3<String, Set<String>, JedisSentinelPoolFeinno>(masterName, sentinels,
                    instance));
        }

        private static void checkSentinel() {
            while (true) {
                try {
                    for (Combo3<String, Set<String>, JedisSentinelPoolFeinno> entry : sentinelGroups) {
                        entry.getV3().refreshMaster(entry.getV1(), entry.getV2());

                    }
                    Thread.sleep(1000);
                } catch (Exception e) {
                    log.error("loop sentinel error", e);
                }
            }
        }
    }

    // protected class MasterListener extends Thread {
    //
    // protected String masterName;
    // protected String host;
    // protected int port;
    // protected long subscribeRetryWaitTimeMillis = 5000;
    // protected Jedis j;
    // protected AtomicBoolean running = new AtomicBoolean(false);
    //
    // protected MasterListener() {
    // }
    //
    // public MasterListener(String masterName, String host, int port) {
    // this.masterName = masterName;
    // this.host = host;
    // this.port = port;
    // }
    //
    // public MasterListener(String masterName, String host, int port, long
    // subscribeRetryWaitTimeMillis) {
    // this(masterName, host, port);
    // this.subscribeRetryWaitTimeMillis = subscribeRetryWaitTimeMillis;
    // }
    //
    // public void run() {
    //
    // running.set(true);
    //
    // while (running.get()) {
    //
    // j = new Jedis(host, port);
    //
    // try {
    //
    // j.subscribe(new JedisPubSub() {
    // @Override
    // public void onMessage(String channel, String message) {
    // log.fine("Sentinel " + host + ":" + port + " published: " + message +
    // ".");
    //
    // String[] switchMasterMsg = message.split(" ");
    //
    // if (switchMasterMsg.length > 3) {
    //
    // if (masterName.equals(switchMasterMsg[0])) {
    // initPool(toHostAndPort(Arrays.asList(switchMasterMsg[3],
    // switchMasterMsg[4])));
    // } else {
    // log.fine("Ignoring message on +switch-master for master name " +
    // switchMasterMsg[0]
    // + ", our master name is " + masterName);
    // }
    //
    // } else {
    // log.severe("Invalid message received on Sentinel " + host + ":" + port
    // + " on channel +switch-master: " + message);
    // }
    // }
    //
    // @Override
    // public void onPMessage(String pattern, String channel, String message) {
    // // TODO Auto-generated method stub
    //
    // }
    //
    // @Override
    // public void onSubscribe(String channel, int subscribedChannels) {
    // // TODO Auto-generated method stub
    //
    // }
    //
    // @Override
    // public void onUnsubscribe(String channel, int subscribedChannels) {
    // // TODO Auto-generated method stub
    //
    // }
    //
    // @Override
    // public void onPUnsubscribe(String pattern, int subscribedChannels) {
    // // TODO Auto-generated method stub
    //
    // }
    //
    // @Override
    // public void onPSubscribe(String pattern, int subscribedChannels) {
    // // TODO Auto-generated method stub
    // }
    // }, "+switch-master");
    //
    // } catch (JedisConnectionException e) {
    //
    // if (running.get()) {
    // log.severe("Lost connection to Sentinel at " + host + ":" + port
    // + ". Sleeping 5000ms and retrying.");
    // try {
    // Thread.sleep(subscribeRetryWaitTimeMillis);
    // } catch (InterruptedException e1) {
    // e1.printStackTrace();
    // }
    // } else {
    // log.fine("Unsubscribing from Sentinel at " + host + ":" + port);
    // }
    // }
    // }
    // }
    //
    // public void shutdown() {
    // try {
    // log.fine("Shutting down listener on " + host + ":" + port);
    // running.set(false);
    // // This isn't good, the Jedis object is not thread safe
    // j.disconnect();
    // } catch (Exception e) {
    // log.severe("Caught exception while shutting down: " + e.getMessage());
    // }
    // }
    // }

}

